package com.atguigu.flink.java.chapter_7;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SessionWindowTimeGapExtractor;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.Date;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2020/12/11 22:42
 */
public class Flink03_chapter07_SessionWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        env
          .socketTextStream("hadoop162", 9999)
          .flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
              @Override
              public void flatMap(String value, Collector<Tuple2<String, Long>> out) throws Exception {
                  Arrays.stream(value.split("\\W+")).forEach(word -> out.collect(Tuple2.of(word, 1L)));
              }
          })
          .keyBy(t -> t.f0)
//          .window(ProcessingTimeSessionWindows.withGap(Time.seconds(10)))
          .window(ProcessingTimeSessionWindows.withDynamicGap(new SessionWindowTimeGapExtractor<Tuple2<String, Long>>() {
              @Override
              public long extract(Tuple2<String, Long> element) { // 返回 gap值, 单位毫秒
                  return element.f0.length() * 1000;
              }
          }))
          .process(new ProcessWindowFunction<Tuple2<String, Long>, Tuple2<String, Long>, String, TimeWindow>() {
              @Override
              public void process(String s, Context context, Iterable<Tuple2<String, Long>> elements, Collector<Tuple2<String, Long>> out) throws Exception {
                  TimeWindow window = context.window();
                  System.out.println(new Date(window.getStart()) + "  " + new Date(window.getEnd()));
              }
          })
          .print();

        env.execute();
    }
}
